package main

import (
	"database/sql"
	"flag"
	"fmt"
	"log"
	"spark-metrics-collector/command"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

var (
	dbUsername = "root"
	dbPassword = "root"
	dbHostname = "localhost"
	dbPort     = "3370"
	dbName     = "resource"
	tableName  = "spark_executor_metrics"
	interval   = 5 // 采集间隔时间（秒）
	url		   = ""
	appId	   = ""
	pid	   	   = ""
	jstat	   = "jstat"
	id	       = "driver"
)

func main() {
	var url string
	flag.StringVar(&dbUsername, "username", "root", "数据库用户名")
	flag.StringVar(&dbPassword, "password", "root", "数据库密码")
	flag.StringVar(&dbHostname, "host", "localhost", "数据库主机名")
	flag.StringVar(&dbPort, "port", "3370", "数据库端口")
	flag.StringVar(&dbName, "db", "resource", "数据库名称")
	// flag.StringVar(&tableName, "table", "spark_executor_metrics", "表名")
	flag.IntVar(&interval, "interval", 5, "采集间隔时间（秒）")
	flag.StringVar(&url, "url", "", "收集的Spark UI地址(yarn webui地址)")
	flag.StringVar(&appId, "appId", "", "Spark的yarn任务ID")
	flag.StringVar(&pid, "pid", "", "jstat监控的本地进程号")
	flag.StringVar(&jstat, "jstat", "jstat", "jstat命令")
	flag.StringVar(&id, "id", "driver", "监听的执行器ID")

	flag.Parse()
	log.Println("request url: " + url)
	if url == "" && pid == "" {
		flag.Usage()
		return
	}

	// 创建数据库连接
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", dbUsername, dbPassword, dbHostname, dbPort, dbName)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()


	if url != "" {
		log.Println("App ID:", appId)
	} else {
		log.Println("Ignore watch app status.")
	}
	// http://merit221.com:8088/proxy/application_1688436482257_0003/api/v1/applications/application_1688436482257_0003/allexecutors
	fullUrl := fmt.Sprintf("%s/proxy/%s/api/v1/applications/%s/allexecutors", url, appId, appId)
	log.Println("app url: " + fullUrl)

	command.SetJstat(jstat)
	log.Println("monitor ID: " + id)

	// 定时采集数据
	ticker := time.NewTicker(time.Duration(interval) * time.Second)
	for range ticker.C {
		if url != "" {
			command.ExecuteSparkCommand(fullUrl, appId, db)
		}
		if pid != "" {
			command.ExecuteStatCommand(pid, appId, id, db)
		}

	}
}


